package com.darcytech.debezium.server.configuration;

import com.darcytech.debezium.common.secret.vault.VaultClient;
import com.darcytech.debezium.core.api.impl.ApiImpl;
import com.darcytech.debezium.core.pool.ConnectorConnectionBinding;
import com.darcytech.debezium.core.rpc.BootStrap;
import com.darcytech.debezium.core.work.VaultHelper;
import com.darcytech.debezium.server.DebeziumProperties;
import com.darcytech.debezium.server.VaultProperties;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class CCBConfiguration {

    private final DebeziumProperties debeziumProperties;

    private final VaultProperties vaultProperties;

    private static final String BASE_LEADER_PATH = "/kafka/connect/manager/";

    public CCBConfiguration(DebeziumProperties debeziumProperties, VaultProperties vaultProperties) {
        this.debeziumProperties = debeziumProperties;
        this.vaultProperties = vaultProperties;
    }

    @Bean
    public VaultHelper vaultHelper() {
        return new VaultHelper(new VaultClient(vaultProperties.getAddress(), vaultProperties.getRoleId(),
                vaultProperties.getSecretId(), vaultProperties.getBasePath()));
    }

    @Bean
    public ConnectorConnectionBinding connectorConnectionBinding() {
        return new ConnectorConnectionBinding(debeziumProperties.getKafkaServers(), debeziumProperties.getKafkaConnectAddresses(), vaultHelper());
    }

    @Bean
    public ApiImpl api() {
        return new ApiImpl();
    }

    @Bean(initMethod = "start")
    public BootStrap bootStrap() {
        return new BootStrap(debeziumProperties.getZkServers(), BASE_LEADER_PATH + debeziumProperties.getKafkaConnectAddresses(),
                debeziumProperties.getRpcAddress(), connectorConnectionBinding(), api());
    }

}